package com.shujia.sql

import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Demo7Save {
  def main(args: Array[String]): Unit = {


    val spark: SparkSession = SparkSession
      .builder()
      .appName("spark")
      .master("local")
      .getOrCreate()


    import spark.implicits._

    val csvDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",") //数据分割方式，默认逗号分割
      //.option("header", "true") //第一行作为列名
      .schema("id STRING, name STRING , age INT , gender STRING ,clazz STRING") //指定表结构
      .load("spark/data/students.txt")


    /**
      *
      * 1、保存为CSV格式的文件
      *
      */

    /**
      *
      * SaveMode
      * Append : 追加
      * Overwrite ： 覆盖
      * ErrorIfExists ： 存在就i报错，默认
      * Ignore ： 存在就忽略
      *
      */

    csvDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save("spark/data/csvout")

    /**
      *
      * 2 保存为json格式
      *
      */
    csvDF.write
      .format("json")
      .mode(SaveMode.Overwrite)
      .save("spark/data/jsonout")


    /**
      *
      * 3 保存parquet格式
      * 会对数据进行压缩，
      * 完全兼容hive
      * 数据中自带列名
      *
      * 解压和压缩需要时间
      *
      */

    csvDF.write
      .format("parquet")
      .mode(SaveMode.Overwrite)
      .save("spark/data/parquetout")

    /**
      * 读取parquet格式
      *
      * 不需要指定列名
      */

    spark
      .read
      .format("parquet")
      .load("spark/data/parquetout")
      .show(1000)


    /**
      *
      * 将数据保存到mysql中
      *
      * createTableColumnTypes : 指定列的类型
      *
      */

    csvDF.write
      .format("jdbc")
      .option("createTableColumnTypes", "id varchar(255), name varchar(255)")
      .option("url", "jdbc:mysql://master:3306?useUnicode=true&characterEncoding=UTF-8")
      .option("dbtable", "student.stu")
      .mode(SaveMode.Overwrite)
      .option("user", "root")
      .option("password", "123456")
      .save()


  }

}
